package com.lgj.rocketmq.product;

import com.alibaba.fastjson.JSONObject;
import com.lgj.rocketmq.code.ProducerTransactionFactory;
import com.lgj.rocketmq.message.RocketMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

/**
 * @author: guangjie.liao
 * @Date: 2023/8/28 22:15
 * @Description: 生产者半消息监听
 */
@Slf4j
public class ProducerTransactionListener implements TransactionListener {

    private ProducerTransactionFactory productLocalMethodHandler;

    public ProducerTransactionListener(ProducerTransactionFactory productLocalMethodHandler){
        this.productLocalMethodHandler = productLocalMethodHandler;
    }


    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        LocalTransactionState state = null;
        try {
            RocketMessage message = JSONObject.parseObject(msg.getBody(),RocketMessage.class);
            productLocalMethodHandler.invokeMethod(message);
            state = LocalTransactionState.COMMIT_MESSAGE;
        } catch (Exception e) {
            log.error("生产者处理事务消息异常：{}",e.getMessage());
            state = LocalTransactionState.ROLLBACK_MESSAGE;
        }
        return state;
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        return null;
    }
}
